package org.xq.softcup.mq.admin.storage;

import com.alibaba.fastjson.JSON;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.xerial.snappy.Snappy;
import org.xq.softcup.mq.client.message.MqMessage;
import org.xq.softcup.mq.util.FileUtil;
import sun.misc.Cleaner;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 文件对应的存储对象
 */
@Slf4j
@Data
public class MqStorage {
    private String basePath; //上级目录
    private String fileName; //文件名
    private String indexName; //索引文件名称
    private AtomicLong mqOffset = new AtomicLong(0l);   //当前已存储的文件量
    private final static long MSG_MAX_SIZE = 1024 * 1024 * 1024L; //文件大小, 默认开1G空间
    private final static long INDEX_FILE_SIZE = 128 * 1024 * 1024; //索引文件大小, 默认128MB
    private FileHolder mqStore;   //mq消息日志
    //各topic 索引存储文件对象
    private Map<String, IndexFileHolder> indexMap =new HashMap<>();
    //文件读写工具类
    private FileUtil fileUtil;

    private ReentrantLock lock = new ReentrantLock();
    private ReentrantLock writeLock = new ReentrantLock();

    /**
     * 获取当前存储对象的全路径, 当前方法要检查文件容量是否已满，如果满了再创建一个文件
     * @return
     */
    public String getMqPath() {
        return basePath + "/" + fileName;
    }

    /**
     * 获取存储消息对象和初始化消息存储对象
     * @return
     */
    public synchronized FileHolder getMqFileHolder() {
        if (mqStore == null) {
            fileUtil = new FileUtil();
            mqStore = new FileHolder();
            mqStore.setPath(basePath);
            //创建缓存
            checkAndCreateFile(mqOffset.get());
        }
        return mqStore;
    }

    private FileHolder checkAndCreateFile(long size) {
        try {
            //消息日志文件名
            String fileName = "mq_storage_" + (int)(size / MSG_MAX_SIZE) + ".log";
            mqStore.setName(fileName);
            if (mqStore.raf != null ) {
                //关闭FileChannel
                mqStore.close();
            }
            RandomAccessFile randomAccessFile = fileUtil.getRandomAccessFile(mqStore.getFullPath());
            mqStore.setRaf(randomAccessFile);
            MappedByteBuffer mappedByteBuffer = randomAccessFile.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, MSG_MAX_SIZE);
            mqStore.setMappedByteBuffer(mappedByteBuffer);
            mqOffset.set(0l); //重置消息读取位置
            return mqStore;
        } catch (IOException e) {
            e.printStackTrace();
            return mqStore;
        }
    }

    /**
     * 根据主题获取索引存储对象，包含初始化索引存储对象
     * @param topic 主题
     * @return
     */
    public synchronized IndexFileHolder getIndexFileHolder(String topic) {
        IndexFileHolder fileHolder = indexMap.get(topic);
        if (fileHolder == null) {
            try {
                fileHolder = new IndexFileHolder();
                fileHolder.setPath(basePath + File.separator + topic);
                String fileName = String.format("%s_index.log", topic);
                fileHolder.setName(fileName);
                //检查索引文件目录
                checkIndexPath(fileHolder.getPath());
                RandomAccessFile randomAccessFile = fileUtil.getRandomAccessFile(fileHolder.getFullPath());
                MappedByteBuffer mappedByteBuffer = randomAccessFile.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, INDEX_FILE_SIZE);
                fileHolder.setMappedByteBuffer(mappedByteBuffer);
                indexMap.put(topic, fileHolder); //添加map里
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return fileHolder;
    }
    /**
     * 检查文件路径，
     * @param path
     * @return
     */
    public String checkIndexPath(String path) {
        File file  = new File(path);
        if (!file.exists()) {
            file.mkdirs();
        }
        return path;
    }


    /**
     * 添加索引到队列，并持久化到本地
     * @param index
     */
    public void addIndex(MqIndex index) {
        try {
//            queue.add(index); //添加道队列
            //获取索引文件对象
            IndexFileHolder indexHolder =  getIndexFileHolder(index.getTopic());
            //添加索引下标
            indexHolder.addIndex(index);
            String data = JSON.toJSONString(index) +",";
            //本地存储
            AtomicLong write = fileUtil.write(indexHolder.getMappedByteBuffer(), data.getBytes("utf-8"), (int) indexHolder.getOffset());
            //保存当前写入的位置
            indexHolder.setOffset(write.get());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 消息保存到文件中
     * @param msgList
     */
    public synchronized long pushMessages(String topic, List<MqMessage> msgList) {
        if (msgList == null || msgList.size() ==0) {
            return 0;
        }
        try {
            //转化为JSON 格式字符串
            String msg = JSON.toJSONString(msgList);
            byte[] compressData = Snappy.compress(msg.getBytes("utf-8"));
            //写入消息日志文件中，返回当前写入的位子
            AtomicLong pos = fileUtil.write(mqStore.getMappedByteBuffer(), compressData, mqOffset.intValue()); //fileUtil.fileWrite(getMqPath(), msg, mqOffset.intValue());
            if (pos.get() > mqOffset.get()) {
                AtomicLong posL = new AtomicLong();
                posL.addAndGet(pos.get() - mqOffset.get());
                log.info("写入: offset={} , size={}", mqOffset.get(), posL);
                //记录当前消息保存的起始位置和长度
                addIndex(new MqIndex(topic, mqOffset.get(), posL.get()));
                // 写入的开始和结束位
                //更新消息的写入位置
                mqOffset.set(pos.get());

                //预测下次文件写入是否超文件大小本身
                checkStorageSize(mqOffset.get()+compressData.length);
            }
            return pos.get();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }
        return 0l;
    }

    /**
     * 检查文件剩余容量，如果容量超了，领创建一个文件
     * @param size
     * @return
     */
    public boolean checkStorageSize(long size) {
        //如果消息的剩余文件大于文件的大小
        if (size >= MSG_MAX_SIZE) {
            checkAndCreateFile(size);
            return true;
        }
        return false;
    }

    /**
     * 根据topic获取消息
     * @param topic
     * @return
     */
    public String popMessage(String topic, String group) {
        try {
//            lock.lock();
//            MqIndex mqPos = queue.pop();
            IndexFileHolder fileHolder = indexMap.get(topic);
            if (fileHolder != null) {
                //获得索引下标
                MqIndex mqPos = fileHolder.popIndex();
                if (mqPos != null) {
                    log.info("读取: offset={} , size={}", mqPos.getOffset(), mqPos.getSize());
                    String msg = fileUtil.read(mqStore.getMappedByteBuffer(), mqPos.getOffset(), mqPos.getSize());//fileUtil.fileRead(getMqPath(), mqPos.getOffset(), mqPos.getSize());
                    return msg;
                }
            }
        } catch (NoSuchElementException e) {
            log.debug("队列无数据可取！");
            return null;
        } catch (IOException e) {//| InterruptedException
            e.printStackTrace();
            log.error("文件读取失败");
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//            log.error("打断线程堵塞");
        } finally {
//            lock.unlock();
        }
        return null;
    }

    /**
     * 批量获取消息
     * @param topic
     * @param size
     * @return
     */
    public List<String> getMsgList(String topic, int size) {
        try {
            List<String> msgList = new ArrayList<>(size);
            while (size > 0) {
//                MqIndex mqPos = queue.pop();
                IndexFileHolder indexHolder = getIndexFileHolder(topic);
                //获得索引下标
                MqIndex mqPos = indexHolder.getIndexQueue().pop();
                String msg = fileUtil.fileRead(indexHolder.getFullPath(), mqPos.getOffset(), mqPos.getSize());
//                MqMessage mqMsg = JSON.parseObject(msg, MqMessage.class);
                size--;
                msgList.add(msg) ;
            }
            return msgList;
        } catch (NoSuchElementException e) {
            log.debug("队列无数据可取！");
        } catch (IOException e) {
            e.printStackTrace();
            log.error("文件读取失败");
        }
        return null;
    }


    /**
     * 文件存储类
     */
    @Data
    class FileHolder {
        private String path;
        private String name;
        private long offset; //偏移量
        private RandomAccessFile raf;
        private MappedByteBuffer mappedByteBuffer; //mmap映射对象
        //获取文件路径
        public String getFullPath() {
            String fullPath = path + File.separator + name;
            File file = new File(fullPath);
            return file.getPath();
        }

        public MappedByteBuffer getMappedBuffer(String tag) {
            if (!tag.equals(name)) {

            }
            return mappedByteBuffer;
        }

        public void close () {
            if (mappedByteBuffer != null) {
                mappedByteBuffer.force();
//                Cleaner cleaner = ((sun.nio.ch.DirectBuffer) mappedByteBuffer).cleaner();
//                if (cleaner != null) {
//                    cleaner.clean();
//                }
                mappedByteBuffer = null;
            }
            if (raf != null) {
                try {
                    //关闭filechannel
                    raf.getChannel().close();
                    raf.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                raf = null;
            }
        }
    }
    /**
     * 索引文件存储类
     */
    @Data
    class IndexFileHolder {
        //索引队列, 每个topic都有
        private LinkedBlockingDeque<MqIndex> indexQueue = new LinkedBlockingDeque();
        private String path;
        private String name;
        private AtomicLong offset = new AtomicLong(0l); //偏移量
        private MappedByteBuffer mappedByteBuffer; //mmap映射对象
        //获取文件路径
        public String getFullPath() {
            String fullPath = path + File.separator + name;
            File file = new File(fullPath);
            return file.getPath();
        }
        public long getOffset() {
            return offset.get();
        }
        public void setOffset(long offset) {
            this.offset.set(offset);
        }
        /**
         * 添加下标对象
         * @param index
         */
        public void addIndex(MqIndex index) {
            indexQueue.add(index);
        }

        public MqIndex popIndex() {
            return indexQueue.pop();
        }
    }
}
